package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/2/5
 *
 * 定义时间属性:
 * https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/concepts/time_attributes/
 *
 *      生成eventtime: 先有水印，再调用  rowtime()
 *      生成processingtime: 调用  proctime()
 *
 *
 */
public class Demo9_DefineTime
{
    public static void main(String[] args) {
        
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
       StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds = env
             .socketTextStream("hadoop103", 8888)
             .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);


        /*
                只选取流中数据的某些属性，组成Table
         */
        Table table = tableEnvironment.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),$("et").rowtime()
            );

        table.execute().print();



    }
}
